-module(rtps_writer).

%%% @doc RTPS Endpoint representing the objects that can be the
%%% sources of messages communicating CacheChanges.

-behaviour(gen_statem).

%% -include_lib("kernel/include/logger.hrl").
-include("rtps.hrl").
-include("rtps_history_cache.hrl").
-include("rtps_sedp.hrl").

%% API
-export([start_link/4,
	 new_change/4, topic_kind/1,
	 add_change/2, remove_change/2,
	 %% Table 8.50 Stateless writer operations
	 reader_locator_add/2, reader_locator_add/3, reader_locator_remove/2,
	 unsent_changes_reset/1,
	 %% Table 8.54 Statefull writer operations
	 matched_reader_add/2, matched_reader_add/3, matched_reader_remove/2,
	 is_acked_by_all/2,
	 set_qos/2,
	 info/1]).

%% gen_statem callbacks
-export([init/1, callback_mode/0,
	 stateless/3, 
	 stateful/3, 
	 handle_event/4, terminate/3, code_change/4]).

-record(data, {sup :: pid(), sedp :: pid() | undefined, opts = [] :: [proplists:property()],
	       guid :: rtps:guid(), writer_cache :: pid() | undefined,
	       u_locs :: [rtps:locator()] | undefined, m_locs :: [rtps:locator()],
	       topic_kind :: rtps:topic_kind(), qos :: undefined | rtps_qos:qos(),
	       last_change_sequence_number :: rtps:sequence_number() | undefined,
	       reg :: pid() | undefined, locators_sup :: pid() | undefined,
	       proxies_sup :: pid() | undefined}).

%%%===================================================================
%%% API
%%%===================================================================

start_link(Sup, Entity_id, Reg, Opts) ->
    gen_statem:start_link({via, rtps_reg, {Reg, Entity_id}}, ?MODULE, [Sup, Opts], []).

%% TODO: add API function(s) to change things like resend_data_period.

-spec new_change(Pid, Kind, Data_value, Instance_handle) -> {ok, Change} | {error, Reason} when
      Pid :: pid(),
      Kind :: rtps_history_cache:change_kind(),
      Data_value :: rtps_history_cache:data_value() | undefined,
      Instance_handle :: rtps_history_cache:instance_handle() | undefined,
      Change :: rtps_history_cache:cache_change(),
      Reason :: ?OUT_OF_RESOURCES.
%% @doc This operation creates a new CacheChange to be appended to the
%% RTPS Writer’s HistoryCache. The sequence number of the CacheChange
%% is automatically set to be the sequenceNumber of the previous
%% change plus one.
%%
%% TODO: There seems to be a problem in the API because according to
%% Figure 8.6 - DDS DataWriter additions to the HistoryCache, it
%% should be possible to use both data and handle when adding a
%% 'alive' cache change. However, on encoding and decoding a data
%% message, the combination of data and key are not allowed and
%% therefor the combination of data and handle was explicitly denied
%% in the API. It is totally unclear how to handle this. It seems
%% logical to allow the handle for a active cache change because the
%% handle defines the key to be used to retrieve the cache change from
%% the cache.
new_change(Pid, Kind, undefined, Instance_handle) ->
    new_change(Pid, Kind, undefined, Instance_handle, byte_size(Instance_handle));
new_change(Pid, Kind, Data_value, Instance_handle) ->
    new_change(Pid, Kind, Data_value, Instance_handle, byte_size(Data_value)).

new_change(Pid, Kind, Data_value, Instance_handle, Size) ->
    case gen_statem:call(Pid, new_change) of
	{ok, {Guid, N}} ->
	    {ok, #cache_change{writer_guid = Guid, sequence_number = N, kind = Kind,
			       data_value = Data_value, instance_handle = Instance_handle,
			       size = Size}};
	{error, _} = Error ->
	    Error
    end.

-spec topic_kind(Pid) -> Topic_kind when
      Pid :: pid(),
      Topic_kind :: rtps:topic_kind().
%% @doc This operation return the topic kind of the writer (no_key / with_key).
topic_kind(Pid) ->
    gen_statem:call(Pid, topic_kind).

-spec add_change(Pid, Change | [Change]) -> ok | {error, Reason} when
      Pid :: pid(),
      Change :: rtps_history_cache:cache_change(),
      Reason :: term().
%% @doc This operation inserts the CacheChange Change into the HistoryCache. 
%% @see rtps_history_cache:add_change/2.
%% NB: This function does not belong to the writer's specs.
add_change(_Pid, []) ->
    ok;
add_change(Pid, Changes)
  when is_list(Changes) ->
    {ok, Writer_cache} = gen_statem:call(Pid, writer_cache),
    rtps_history_cache:add_change(Writer_cache, Changes),
    Seq_nums = [Seq_num || #cache_change{sequence_number = Seq_num} <- Changes],
    gen_statem:call(Pid, {add_change, Seq_nums});
add_change(Pid, Change) ->
    add_change(Pid, [Change]).

-spec remove_change(Pid, Change) -> ok when
      Pid :: pid(),
      Change :: rtps_history_cache:cache_change().
%% @doc This operation indicates that a previously-added CacheChange
%% has become irrelevant and the details regarding the CacheChange
%% need not be maintained in the HistoryCache.
%% NB: This function does not belong to the writer's specs.
remove_change(Pid, Change) ->
    {ok, Writer_cache} = gen_statem:call(Pid, writer_cache),
    rtps_history_cache:remove_change(Writer_cache, Change).
    
-spec reader_locator_add(Pid, Locator) -> Result when
      Pid :: pid(),
      Locator :: rtps:locator(),
      Result :: supervisor:startchild_ret() | {error, Reason},
      Reason :: term(). 
%% @doc This operation adds the Locator `Locator' to the StatelessWriter::reader_locators.
reader_locator_add(Pid, Locator) ->
    reader_locator_add(Pid, Locator, []).

-spec reader_locator_add(Pid, Locator, Opts) -> Result when
      Pid :: pid(),
      Locator :: rtps:locator(),
      Opts :: [proplists:property()],
      Result :: supervisor:startchild_ret() | {error, Reason},
      Reason :: term(). 
%% @doc This operation adds the Locator `Locator' to the StatelessWriter::reader_locators.
reader_locator_add(Pid, Locator, Opts) ->
    gen_statem:call(Pid, {reader_locator_add, Locator, Opts}).

-spec reader_locator_remove(Pid, Locator) -> ok when
      Pid :: pid(),
      Locator :: rtps:locator().
%% @doc This operation removes the Locator `Locator' from the StatelessWriter::reader_locators.
reader_locator_remove(Pid, Locator) ->
    gen_statem:cast(Pid, {reader_locator_remove, Locator}).

-spec unsent_changes_reset(Pid) -> ok when
      Pid :: pid().
%% @doc This operation modifies the set of unsent_changes for all
%% the ReaderLocators in the StatelessWriter::reader_locators.  The
%% list of unsent changes is reset to match the complete list of
%% changes available in the writer’s HistoryCache.
unsent_changes_reset(Pid) ->
    gen_statem:call(Pid, unsent_changes_reset).
    
-spec matched_reader_add(Pid, Reader) -> Result when
      Pid :: pid(),
      Reader :: rtps:guid(),
      Result :: supervisor:startchild_ret() | {error, Reason},
      Reason :: term(). 
%% @doc This operation adds the ReaderProxy GUID `Reader' to the set
%% StatefulWriter::matched_readers.
matched_reader_add(Pid, Reader) ->
    matched_reader_add(Pid, Reader, []).

-spec matched_reader_add(Pid, Reader, Opts) -> Result when
      Pid :: pid(),
      Reader :: rtps:guid(),
      Opts :: [proplists:property()],
      Result :: supervisor:startchild_ret() | {error, Reason},
      Reason :: term(). 
%% @doc This operation adds the ReaderProxy GUID `Reader' to the set
%% StatefulWriter::matched_readers.
matched_reader_add(Pid, Reader, Opts) ->
    gen_statem:call(Pid, {matched_reader_add, Reader, Opts}).

-spec matched_reader_remove(Pid, Reader) -> ok when
      Pid :: pid(),
      Reader :: rtps:guid().
%% @doc This operation removes the ReaderProxy GUID `Reader' from the
%% set StatefulWriter::matched_readers.
matched_reader_remove(Pid, Reader) ->
    gen_statem:cast(Pid, {matched_reader_remove, Reader}).

-spec is_acked_by_all(Pid, Change) -> true | false when
      Pid :: pid(),
      Change :: rtps_history_cache:cache_change().
%% @doc This operation takes a CacheChange `Change' as a parameter and
%% determines whether all the ReaderProxy have acknowledged the
%% CacheChange. The operation will return true if all ReaderProxy have
%% acknowledged the corresponding CacheChange and false otherwise.
is_acked_by_all(Pid, #cache_change{sequence_number = Seq_num}) ->
    gen_statem:call(Pid, {is_acked_by_all, Seq_num}).

-spec set_qos(Pid, Qos) -> ok when
    Pid :: pid(),
    Qos :: [rtps:parameter()].
%% @doc This operation is used to set the QoS policies of the
%% writer. The set of policies specified as the qos_list parameter
%% replace the existing QoS, This operation will not perform any
%% checks, which is left for the external application to do.
set_qos(Pid, Qos)
  when is_list(Qos) ->
    gen_statem:call(Pid, {set_qos, Qos}).

info(Pid) ->
    gen_statem:call(Pid, info).

%%%===================================================================
%%% gen_statem callbacks
%%%===================================================================

%% @private
init([Sup, Opts]) ->
    State_type = proplists:get_value(state_type, Opts),
    Guid_prefix = proplists:get_value(guid_prefix, Opts),
    Entity_id = proplists:get_value(entity_id, Opts),
    Guid = #guid{guid_prefix = Guid_prefix, entity_id = Entity_id},
    U_locs = proplists:get_value(uniicast_locator_list, Opts),
    M_locs = proplists:get_value(multicast_locator_list, Opts),
    Topic_kind = proplists:get_value(topic_kind, Opts),
    Writer_cache = proplists:get_value(history_cache, Opts),
    Qos = proplists:get_value(qos, Opts, []),
    case Entity_id of
	#entity_id{type = user} ->
	    %% Only insert endpoints in the SEDP which are of type
	    %% `user'. It may be also correct to insert endpoints of
	    %% type `vendor' and `unknown', but never `built_in' or
	    %% one of the endpoints with a predefined entity ids.
	    process_flag(trap_exit, true),
	    Sedp = proplists:get_value(sedp, Opts),
	    Writer_proxy = #writer_proxy{remote_writer_guid = Guid,
					 unicast_locator_list = U_locs,
					 multicast_locator_list = M_locs},
	    ok = rtps_sedp:insert(Sedp, Writer_proxy, Qos);
	_ ->
	    Sedp = undefined
    end,
    Data = #data{sup = Sup, sedp = Sedp, opts = Opts, guid = Guid,
		 u_locs = U_locs, m_locs = M_locs, qos = Qos,
		 topic_kind = Topic_kind, writer_cache = Writer_cache},
    {ok, State_type, Data}.

%% @private
callback_mode() ->
    [state_functions, state_enter].

%% @private
stateless(enter, _Old_state,
	  #data{sup = Sup, opts = Opts, guid = Guid, u_locs = U_locs, m_locs = M_locs,
		writer_cache = Writer_cache} = Data) ->
    Writer_cache1 = writer_cache(Sup, Writer_cache),
    Children = supervisor:which_children(Sup),
    {ok, Reg} = child(Children, registry),
    {ok, Locators_sup} = child(Children, locators),
    Locs = case U_locs of
	       undefined -> M_locs;
	       _ -> M_locs ++ U_locs
	   end,
    [add_locator(Locators_sup, Reg, Locator, Guid, Writer_cache1, Opts) || Locator <- Locs],
    {keep_state, Data#data{writer_cache = Writer_cache1, reg = Reg, locators_sup = Locators_sup}};
stateless({call, From}, {reader_locator_add, Locator, Extra_opts},
	  #data{reg = Reg, locators_sup = Locators_sup, opts = Opts, guid = Guid, writer_cache = Writer_cache}) ->
    Reply = add_locator(Locators_sup, Reg, Locator, Guid, Writer_cache, Extra_opts ++ Opts),
    %% TODO: update the SEDP
    {keep_state_and_data, [{reply, From, Reply}]};
stateless(cast, {reader_locator_remove, Locator}, #data{locators_sup = Locators_sup} = _Data) ->
    rtps_ofo_sup:stop_child(Locators_sup, Locator),
    %% TODO: update the SEDP
    keep_state_and_data;
stateless({call, From}, {add_change, Seq_nums}, #data{reg = Reg, qos = Qos} = _Data) ->
    Locators = rtps_reg:which(Reg),
    [rtps_reader_locator:add_change(Pid, Seq_nums, Qos) || {_Id, Pid} <- Locators],
    {keep_state_and_data, [{reply, From, ok}]};
stateless({call, From}, unsent_changes_reset,
	  #data{reg = Reg, qos = Qos, guid = Guid, writer_cache = Writer_cache} = _Data) ->
    case rtps_history_cache:get_seq_num_min_max(Writer_cache, Guid) of
	{First_sn, Last_sn} ->
	    Locators = rtps_reg:which(Reg),
	    Seq_nums = lists:seq(First_sn, Last_sn),
	    [rtps_reader_locator:add_change(Pid, Seq_nums, Qos) || {_Id, Pid} <- Locators];
	_ ->
	    ok
    end,
    {keep_state_and_data, [{reply, From, ok}]};
stateless(Event_type, Event_content, Data) ->
    handle_event(Event_type, Event_content, stateless, Data).

%% @private
stateful(enter, _Old_state, #data{sup = Sup, writer_cache = Writer_cache} = Data) ->
    Writer_cache1 = writer_cache(Sup, Writer_cache),
    Children = supervisor:which_children(Sup),
    {ok, Reg} = child(Children, registry),
    {ok, Proxies_sup} = child(Children, proxies),
    {keep_state, Data#data{writer_cache = Writer_cache1, reg = Reg, proxies_sup = Proxies_sup}};
stateful({call, From}, {matched_reader_add, Reader, Extra_opts},
	 #data{reg = Reg, proxies_sup = Proxies_sup, opts = Opts, guid = Guid, writer_cache = Writer_cache}) ->
    Spec = #{id => Reader,
	     start => {rtps_reader_proxy, start_link, [Reader, Reg, Guid, Writer_cache, Extra_opts ++ Opts]},
	     restart => permanent,
	     type => worker},
    Reply = rtps_ofo_sup:start_child(Proxies_sup, Spec),
    %% TODO: update the SEDP
    {keep_state_and_data, [{reply, From, Reply}]};
stateful(cast, {matched_reader_remove, Reader}, #data{proxies_sup = Proxies_sup}) ->
    rtps_ofo_sup:stop_child(Proxies_sup, Reader),
    %% TODO: update the SEDP
    keep_state_and_data;
stateful({call, From}, {add_change, Seq_nums}, #data{reg = Reg, qos = Qos} = _Data) ->
    Proxies = rtps_reg:which(Reg),
    [rtps_reader_proxy:add_change(Pid, Seq_nums, Qos) || {_Id, Pid} <- Proxies],
    {keep_state_and_data, [{reply, From, ok}]};
stateful({call, From}, {is_acked_by_all, Seq_num}, #data{reg = Reg} = _Data) ->
    Proxies = rtps_reg:which(Reg),
    Pred = fun(Pid) ->
		   rtps_reader_proxy:is_acked(Pid, Seq_num)
	   end,
    Reply = lists:all(Pred, Proxies),
    {keep_state_and_data, [{reply, From, Reply}]};
stateful(Event_type, Event_content, Data) ->
    handle_event(Event_type, Event_content, stateful, Data).

%% @private
handle_event({call, From}, writer_cache, _State_name, #data{writer_cache = Writer_cache}) ->
    Reply = {ok, Writer_cache},
    {keep_state_and_data, [{reply, From, Reply}]};
handle_event({call, From}, new_change, _State,
	     #data{guid = Guid, last_change_sequence_number = undefined} = Data) ->
    Reply = {ok, {Guid, 1}},
    {keep_state, Data#data{last_change_sequence_number = 1}, [{reply, From, Reply}]};
handle_event({call, From}, new_change, _State,
	     #data{guid = Guid, last_change_sequence_number = Last} = Data)
  when Last < ?MAX_SEQUENCE_NUMBER ->
    Last1 = Last + 1,
    Reply = {ok, {Guid, Last1}},
    {keep_state, Data#data{last_change_sequence_number = Last1}, [{reply, From, Reply}]};
handle_event({call, From}, new_change, _State, _Data) ->
    Reply = {error, ?OUT_OF_RESOURCES},
    {keep_state_and_data, [{reply, From, Reply}]};
handle_event({call, From}, topic_kind, _State, #data{topic_kind = Topic_kind}) ->
    Reply = Topic_kind,
    {keep_state_and_data, [{reply, From, Reply}]};
handle_event({call, From}, {set, Qos}, _State, Data) ->
    %% TODO: update the SEDP
    Reply = ok,
    {keep_state, Data#data{qos = Qos}, [{reply, From, Reply}]};
handle_event({call, From}, info, State_name,
	     #data{reg = Reg, guid = #guid{entity_id = Entity_id}, topic_kind = Topic_kind}) ->
    Children = case State_name of
		   stateless ->
		       {locators, [Locator || {Locator, _Pid} <- rtps_reg:which(Reg)]};
		   stateful ->
		       {proxies, [Proxy || {Proxy, _Pid} <- rtps_reg:which(Reg)]}
	       end,
    Reply = [{entity_id, Entity_id}, {topic_kind, Topic_kind}, Children],
    {keep_state_and_data, [{reply, From, Reply}]};

%% TODO: Sink unhadled events. Remove these clauses.
handle_event({call, From} = _Event_type, _Event_content, _State, _Data) ->
    Reply = ok,
    {keep_state_and_data, [{reply, From, Reply}]};
handle_event(_Event_type,  _Event_content, _State, _Data) ->
    keep_state_and_data.

%% @private
terminate(_Reason, _State, #data{sedp = Sedp, guid = Guid}) ->
    rtps_sedp:delete(Sedp, #writer_proxy{remote_writer_guid = Guid}),
    ok.

%% @private
code_change(_Old_vsn, State, Data, _Extra) ->
    {ok, State, Data}.

%%%===================================================================
%%% Internal functions
%%%===================================================================

%% @private
child(Children, Id) ->
    case lists:keyfind(Id, 1, Children) of
	{_Id, Child, _Type, _Modules} when is_pid(Child) ->
	    {ok, Child};
	{_Id, Child, _Type, _Modules} when Child =:= restarting ->
	    {error, Child};
	_ ->
	    {error, notfound}
    end.

%% @private
%% @doc Add a history cache to the endpoint's supervisor if
%% Writer_cache is not undefined.
writer_cache(Sup, undefined) ->
    {ok, Pid} = rtps_endpoint_sup:history_cache(Sup),
    Pid;
writer_cache(_Sup, Writer_cache) ->
    Writer_cache.

%% @private
%% @doc Add a reader locator to the locators supervisor
add_locator(Locators_sup, Reg, Locator, Guid, Writer_cache, Opts) ->
    Spec = #{id => Locator,
	     start => {rtps_reader_locator, start_link, [Locator, Reg, Guid, Writer_cache, Opts]},
	     restart => permanent,
	     type => worker},
    {ok, _Pid} = rtps_ofo_sup:start_child(Locators_sup, Spec).

